<?php

namespace app\command;


use app\command\extend\http\ProfitStatistics;
use app\common\model\Calculation;
use app\common\model\coin\CallApi;
use app\common\model\coin\Coin;
use app\common\model\coin\CoinRule;
use app\common\model\coin\CoinRuleRun;
use app\common\model\coin\CoinRuleRunTrade;
use app\common\model\system\SqlToRedis;
use app\common\model\users\Users;
use mall\library\Huobi;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Exception;
use swoole_server;
use think\facade\Db;


/**
 * 获取历史火币数据，并进行保存
 * Class HistoryTrade
 * @package app\command
 */
class UdpServer extends Command
{


    /**
     * 进程参数设置
     */
    const PROCESS_CONFIG = [
        'SERVER'    => [
            'REACTOR_NUM'   => 4,   // 最大线程数
            'WORKER_NUM'    => 4,  // 工作进程数
            'TASK_WORKER_NUM'    => 16,  // tash 工作进程数
            'MAX_REQUEST'   => 10000,    // 最大请求数
            'DEFAULT_PORT'  => 13902,   // 默认端口
            'DAEMONIZE' => 0,   //是否守护进程
        ],
        'DEVIATIONS_CORRECT'    => [
            'LOG_TYPE'  => 'DEVIATIONS_CORRECT',  //纠正日志类型
            'RESULT_TYPE'    => [    //纠正结果类型
                'NULL'      => 'null',  //记录为空
                'IGNORE'    => 'ignore',    // 忽视、无需处理
                'CORRECT'   => 'correct',   // 纠正
                'CORRECT_ERROR'   => 'correct_error',   // 纠正失败
                'ERROR'   => 'error',   // 发生错误，例如请求失败
            ],

            'RESTART_CALL_API_TIMES'=> 3,   //请求第三方最大次数
            'RESTART_CALL_API_TIME' => 1,   //请求第三方失败重试时间[单位s]

        ],
    ];

    /**
     * ENV 配置参数KEY列表
     */
    const ENV_KEY_LIST = [
        'SERVER_UDP_IP_LIST'    => 'udp_server.localhost_ip',   // UDP 本机ip地址列
        'SERVER_UDP_PORT'   => 'udp_server.port', // UDP 服务器端口
        'SERVER_UDP_CMD_IP_LIMIT_DEFAULT'  => 'udp_server.cmd_ip_limit_default',    //IP 限制权限【默认限制IP，当对应cmd限制不存在时使用】
        'SERVER_UDP_CMD_IP_LIMIT'  => 'udp_server.cmd_ip_limit',    //IP 限制权限

        'CMD_REQUEST_SYNC_ADDR_DEFAULT' => 'udp_server.cmd_request_sync_addr_default',   // CMD 统一同步地址,
        'CMD_REQUEST_SYNC_ADDR' => 'udp_server.cmd_request_sync_addr',   // 单独 CMD 需同步地址

    ];

    /**
     * cmd 列表
     */
    const CMD_LIST = [
        'CALL_API'  => 'callApi',    // 同步请求第三方
        'BALANCE_CHANGE'  => 'balanceChange',    // 余额变更处理
        'REFRESH'  => 'refresh',    // 刷新数据接口
        'DEVIATIONS_CORRECT'  => 'deviationsCorrect',  // 纠正偏差
        'API_CACHE_CALL'  => 'apiCacheCall',  // api 缓存后台处理
        'VALIDATE'  => 'validate',  //验证允许状态
    ];

    /**
     * CMD 中刷新(REFRESH)数据接口类型列表
     */
    const CMD_REFRESH_TYPE_LIST = [
        'RULE'  => 'rule',  // 刷新规则
        'ORDER' => 'order', // 刷新订单
    ];

    // cmd IP限制
    private $cmdIpLimit = [];

    // cmd IP 默认限制
    private $cmdIpLimitDefault = [];


    /**
     * 重写configure
     * {@inheritdoc}
     */
    protected function configure()
    {
        // 命令的名字（"think" 后面的部分）
        $this->setName('UdpServer')
            // 运行 "php think list" 时的简短描述
            ->setDescription('开启http服务器，监控请求处理')
            // 运行命令时使用 "--help" 选项时的完整命令描述
            ->setHelp("一直执行 无参数");
    }

    /**
     *  * 重写execute
     *  * {@inheritdoc}
     *
     * @param Input $input
     * @param Output $output
     */
    public function execute(Input $input, Output $output)
    {
        //  获取 cmd 中的 IP 限制
        $this->getLimit();

        // 开启监听
        $this->createServer();
    }


    /**
     * 获取 cmd 中的 IP 限制
     */
    private function getLimit() {
        // 获取IP限制
        $limit = trim(env(self::ENV_KEY_LIST['SERVER_UDP_CMD_IP_LIMIT']));
        $limit = explode(' ',$limit);
        array_map(function($limit){
            $limit = explode(':',$limit);
            if(count($limit) != 2) {
                return '';
            }
            $this->cmdIpLimit[strtoupper($limit[0])] = $limit[1] == 0?0:array_map(function($ip){return trim($ip);},explode(',',$limit[1]));
            return '';
        },$limit);

        // 获取默认限制
        $this->cmdIpLimitDefault = explode(',',trim(env(self::ENV_KEY_LIST['SERVER_UDP_CMD_IP_LIMIT_DEFAULT'])));

        return true;
    }

    /**
     * 发送消息
     * @param $message 消息内容，随意值，会直接json处理
     * @param int|null $portIndex
     * @return bool
     */
    static function request(string $cmd,string $ipList = null,array $params = null,int $timeout = 2) {
        /************* 获取同步请求地址 ****************/
        if(empty($ipList)) {
            // 获取本机地址列表
            $localhostIpList = array_filter(explode(',',trim(env(self::ENV_KEY_LIST['SERVER_UDP_IP_LIST']))));
            $localhostIpList = array_unique(array_merge($localhostIpList,['localhost','127.0.0.1']));

            // 获取同步的IP列表
            $syncList = null;
            $cmdUpper = strtoupper($cmd);
            $cmdRequestSyncAddr = explode(' ',trim(env(self::ENV_KEY_LIST['CMD_REQUEST_SYNC_ADDR'])));
            foreach($cmdRequestSyncAddr as $cmdRequestDetails) {
                $cmdRequestDetails = explode(',',$cmdRequestDetails);
                if(strtoupper($cmdRequestDetails[0]) == $cmdUpper) {
                    array_shift($cmdRequestDetails);
                    if(!$syncList = implode(',',$cmdRequestDetails)) {
                        $syncList = 'localhost';
                    }
                    break;
                }
            }

            if(!$syncList) {    // 使用默认IP配置
                $syncList = trim(env(self::ENV_KEY_LIST['CMD_REQUEST_SYNC_ADDR_DEFAULT']));
                $syncList = empty($syncList)?'localhost':$syncList;
            }
            // 获取本机地址列表
            $syncList = array_map(function($addr) use($localhostIpList){
                if(in_array(explode(':',$addr)[0],$localhostIpList)) {
                    $addr = '127.0.0.1:' . self::serverPort();
                }

                return $addr;
            },explode(',',$syncList));
            $syncList = array_unique($syncList);
            $ipList = implode(',',$syncList);
        }
        /*********** END 获取同步请求地址 **************/

        $params = [
            'cmd'   => $cmd,
            'expire'    => microtime(true) + $timeout * 1000,
            'params'    => $params,
        ];
        // 编译参数
        $params = self::encode($params);
        $hostList = array_map(function($host){
            if(!$host = trim($host)) {
                return '';
            }
            $host = explode(':',$host);
            if(count($host) == 1) {
                $host[1] = 80;
            }
            return $host;
        },explode(',',$ipList));
        $hostList = array_filter($hostList);
        foreach($hostList as $host) {
            try {
                $client = new \swoole_client(SWOOLE_SOCK_UDP);
                $client->sendTo($host[0],$host[1],$params);
                $client->close();
            } catch (\Exception $e) {
            }
        }
        return true;
    }


    /**
     * 启动服务器
     */
    private function createServer() {
        // 创建udp服务器
        $server = new swoole_server('0.0.0.0',self::serverPort(),SWOOLE_PROCESS, SWOOLE_SOCK_UDP);
        $server->set([
            'reactor_num'   => self::PROCESS_CONFIG['SERVER']['REACTOR_NUM'],     // reactor thread num
            'worker_num'    => self::PROCESS_CONFIG['SERVER']['WORKER_NUM'],     // worker process num
            'task_worker_num'    => self::PROCESS_CONFIG['SERVER']['TASK_WORKER_NUM'],     // task worker process num
            'max_request'   => self::PROCESS_CONFIG['SERVER']['MAX_REQUEST'],
            'daemonize'   => self::PROCESS_CONFIG['SERVER']['DAEMONIZE'],   //是否开启守护进程
        ]);


        //监听数据接收事件
        $server->on('Packet', function ($serv, $data, $clientInfo) {
            $serv->task($data . ' | ' . $clientInfo['address']);
        });
        $server->on('Task',function($serv, $task_id, $from_id, $data){
            try {
                list($data,$address) = explode(' | ',$data);
                $params = self::decode($data);
                $result = $this->execReuqest($params,$address);
                // 添加执行日志
                Log::write([
                    'params'    => $params,
                    'result'    => $result,
                ],Log::LOG_TYPE_LIST['UDP_API']);
            }catch (\Exception $e) {
                Log::write($data . ' ' . $e->getMessage(),Log::LOG_TYPE_LIST['UDP_API_ERR'],false);
            }
            return true;
        });

        $server->on('Finish',function($serv,$task_id, $data){

        });

        $server->start();

    }

    /**
     * 执行请求，并进行限制处理
     * @param $data 接收的数据
     */
    public function execReuqest($request,$remoteAddr) {
        if(empty($request) || !array_key_exists('cmd',$request) || !array_key_exists('expire',$request) || !array_key_exists('params',$request)) {
            $this->throwErr(0);
        }
        if($request['cmd'] == self::CMD_LIST['VALIDATE']) {
            return Crontab::validateWrite($request['params']);
        }
        // 获取客户端地址
        if(!array_key_exists('cmd',$request) || empty($request['cmd']) || !method_exists($this,($cmdMethod = 'cmd' . strtoupper($request['cmd'][0]) . substr($request['cmd'],1)))) {
            $this->throwErr(-1);
        }
        $cmd = strtoupper($request['cmd']);
        if(!in_array($remoteAddr,['127.0.0.1']) && (
                (array_key_exists($cmd,$this->cmdIpLimit) && $this->cmdIpLimit[$cmd] != 0 && !in_array($remoteAddr,$this->cmdIpLimit[$cmd]))
                ||
                (!array_key_exists($cmd,$this->cmdIpLimit) && !in_array($remoteAddr,$this->cmdIpLimitDefault))
            )

        ) {
            $this->throwErr(-2);
        }

        // 判断有效期
        if(microtime(true) - $request['expire'] > 0) {
            $this->throwErr(-3);
        }
        $result = [
            'status'    => 'ok',
            'errMsg'    => '',
            'data'      => null,
        ];
        try {
            $result['data'] = call_user_func_array([$this,$cmdMethod],[$request['params']]);
        }catch (\Exception $e) {
            $result['status'] = 'err';
            $result['errMsg'] = $e->getMessage();
        }
        Db::close();
        return $result;
    }


    /**
     * 调用火币 API 接口
     * @param array $params
     * @throws Exception
     */
    private function cmdCallApi(array $params) {
        $result = (new Huobi())->curl($params['url'],$params['params']);
        if(empty($result) || !($result = json_decode($result,true))) {
            $this->throwErr(1);
        }
        // 返回执行结果
        return $result;
    }

    /**
     * 余额变更处理
     * @param array $params
     */
    private function cmdBalanceChange(array $params) {
        // 验证参数
        if(!array_key_exists('userId',$params) || !array_key_exists('apiStatus',$params) || !array_key_exists('usdt',$params) || !array_key_exists('changeList',$params)) {
            $this->throwErr(0); // 返回参数错误
        }
        $userId = intval($params['userId']);
        // 修改账户表信息
        $where = [
            'id'    => $userId,
        ];
        $updateData = [];
        if($params['apiStatus'] == 'error') {  // api 请求失败处理
            $where['amount_query_err_time'] = 0;
            $updateData['amount_query_err_time'] = microtime(true);
        } else {  // api 请求成功，执行修改
            $updateData['amount_query_err_time'] = 0;
            if(in_array('usdt',$params['changeList'])) {
                $updateData['amount'] = $params['usdt'];
            }
        }
        // 修改用户资金状态
        try {
            // 修改用户信息
            (new Users())->where($where)->update($updateData);
        } catch(\Exception $e) {
            // 记录sql错误信息
            Log::write(Swoole::getErrorAndFiles($e),Log::LOG_TYPE_LIST['SQL_ERROR'],false);
        }
        foreach($params['changeList'] as $code) {
            try {
                if($code == 'usdt') {   // 余额变更处理
                    $updateWhere = [
                        ['rule.user_id','=',$userId],
                        ['run.last_buy_err_time','>',0],
                    ];
                    if($ruleIdList = (new CoinRule())->alias('rule')
                        ->join('Coin_rule_run run','rule.coin_rule_run_id=run.id','inner')
                        ->where($updateWhere)
                        ->column('rule.id')) {
                        (new CoinRule())->alias('rule')
                            ->join('Coin_rule_run run','rule.coin_rule_run_id=run.id','inner')
                            ->where($updateWhere)->update(['run.last_buy_err_time'=>0]);

                        // 调用刷新调用规则
                        self::request(self::CMD_LIST['REFRESH'],null,[
                            'type'  => self::CMD_REFRESH_TYPE_LIST['RULE'],
                            'params'    => [
                                'ruleIdList' => $ruleIdList,
                            ],
                        ]);
                    }
                } else {    // 订单变更处理
                    // 判断币种是否存在
                    if($coinList = (new Coin())->where(['code'=>$code . 'usdt'])->column('id')) {
                        foreach($coinList as $coinId) {
                            // 调用刷新订单监控
                            self::request(self::CMD_LIST['REFRESH'],null,[
                                'type'  => self::CMD_REFRESH_TYPE_LIST['ORDER'],
                                'params'    => [
                                    'userId'    => $userId,
                                    'coinId'    => $coinId,
                                ],
                            ]);
                        }
                    }

                }
            } catch(\Exception $e) {
                // 记录sql错误信息
                Log::write(Swoole::getErrorAndFiles($e),Log::LOG_TYPE_LIST['SQL_ERROR'],false);
            }
        }
        return true;
    }

    /**
     * 偏差纠正
     * @param array $params
     */
    private function cmdDeviationsCorrect(array $params) {
        // 验证参数
        if(!array_key_exists('runId',$params)) {
            $this->throwErr(0); // 返回参数错误
        }
        $scale = (new CoinRuleRun())->scale;
        list($del,$whereTradeMonitor) = CoinRuleRunTrade::getStatusAndWhere('MONITOR');
        $logWriteFunc = function($runId,$resultType,$value = 'null') {
            Log::write(implode(' ',[$runId,self::PROCESS_CONFIG['DEVIATIONS_CORRECT']['RESULT_TYPE'][$resultType],$value]),Log::LOG_TYPE_LIST[self::PROCESS_CONFIG['DEVIATIONS_CORRECT']['LOG_TYPE']],false);
        };
        // 获取规则ID
        $runIdList = $params['runId'];
        if(!is_array($runIdList)) {
            $runIdList = [$runIdList];
        }
        foreach($runIdList as $runId) {
            try{
                $runId = intval($runId);
                if(!$runDetails = (new CoinRuleRun())->alias('run')
                    ->join('Coin_rule rule','run.coin_rule_id=rule.id','inner')
                    ->join('Coin coin','coin.id=rule.coin_id','inner')
                    ->join('Users_key key','key.id=rule.user_id','inner')
                    ->where(['run.id'=>$runId])
                    ->field([
                        'run.version','run.quantity','run.field_fees','run.deduct_fees','run.last_remaining_quantity','run.deviations_quantity','last_sell_err_time','run.coin_rule_id',
                        'rule.deviations_calculation','rule.coin_rule_run_id',
                        'key.account_id','key.access_key','key.secret_key',
                        'coin.code',
                    ])
                    ->find()) {
                    // 记录不存在
                    $logWriteFunc($runId,'NULL');
                }

                if($runDetails['coin_rule_run_id'] != $runId || (new CoinRuleRunTrade())->where(array_merge($whereTradeMonitor,[['coin_rule_run_id','=',$runId]]))->count()) {
                    $logWriteFunc($runId,'IGNORE');
                }

                // 获取余额
                for($i = 0;$i<self::PROCESS_CONFIG['DEVIATIONS_CORRECT']['RESTART_CALL_API_TIMES'];++$i) {
                    if($balance = (new Huobi($runDetails['account_id'],$runDetails['access_key'],$runDetails['secret_key']))->get_balance($runDetails['account_id'])) {
                        if($balance['status'] == 'ok') {
                            break;
                        }
                    }
                    sleep(self::PROCESS_CONFIG['DEVIATIONS_CORRECT']['RESTART_CALL_API_TIME']);
                }
                if(!$balance) {
                    $logWriteFunc($runId,'ERROR','network');
                } else if($balance['status'] == 'error') { // 获取成功，整理偏差
                    $logWriteFunc($runId,'ERROR',CallApi::getSaveError($balance));
                } else {
                    // 余额获取成功，计算及保存偏差量
                    $balanceTrade = null;
                    foreach($balance['data']['list'] as $currencyDetails) {
                        if($currencyDetails['currency'] . 'usdt' == $runDetails['code']&& $currencyDetails['type'] == 'trade') {
                            $balanceTrade = $currencyDetails['balance'];
                            break;
                        }
                    }
                    if(is_null($balanceTrade)) {    // 获取余额失败
                        $logWriteFunc($runId,'ERROR','get balance error');
                        break;
                    }

                    // 获取系统保存的总数量
                    $quantity = bcadd($runDetails['quantity'],$runDetails['last_remaining_quantity'],$scale);
                    $quantity = bcadd($quantity,$runDetails['deduct_fees'],$scale);
                    $deviationsQuantity = bcsub($balanceTrade,$quantity,$scale);    // 获取系统偏差量
                    if(bccomp($deviationsQuantity,$runDetails['deviations_quantity'],$scale) != 0 && (bccomp($deviationsQuantity,0,$scale) == -1 || $runDetails['deviations_calculation'] == CoinRule::DEVIATIONS_CALCULATION_TYPE['OPEN'])) {
                        // 进行纠正处理
                        if((new CoinRuleRun())->recordUpdate([
                            'id'    => $runId,
                            'unique_id' => $runDetails['coin_rule_id'],
                            'version'   => $runDetails['version'],
                        ],[
                            'deviations_quantity'=>$deviationsQuantity,
                            'last_sell_err_time'=>0,
                        ])) {
                            $logWriteFunc($runId,'CORRECT',$deviationsQuantity);
                            if($runDetails['last_sell_err_time']) { // 存在卖出错误，通知更新策略
                                // 调用刷新调用规则怕
                                self::request(self::CMD_LIST['REFRESH'],null,[
                                    'type'  => self::CMD_REFRESH_TYPE_LIST['RULE'],
                                    'params'    => [
                                        'ruleIdList' => [$runDetails['coin_rule_id']],
                                    ],
                                ]);
                            }
                        } else {
                            $logWriteFunc($runId,'CORRECT_ERROR',$deviationsQuantity);
                        }
                    } else {
                        $logWriteFunc($runId,'IGNORE',$deviationsQuantity);
                    }
                }
            } catch(\Exception $e) {
                // 记录sql错误信息
                Log::write(Swoole::getErrorAndFiles($e),Log::LOG_TYPE_LIST['SQL_ERROR'],false);
            }
        }

        return true;
    }


    /**
     * 刷新参数
     * @param array $params
     */
    private function cmdRefresh(array $params) {
        switch($params['type']) {
            case self::CMD_REFRESH_TYPE_LIST['RULE']:   // 刷新规则
                for($i=0;$i<3;++$i) {
                    try {
                        HistoryTrade::tcpTableQuery('userRuleRefresh',$i);  //除不使用redis之外，其他直接返回操作结果
                    }catch (\Exception $e) {
                    }
                }
                break;
            case self::CMD_REFRESH_TYPE_LIST['ORDER']:  // 刷新订单 【后续按单独】
                (new SqlToRedis())->getTradeMonitor(true);
                break;
        }
    }

    /**
     * apiCache 对接中转站
     * @param array $params
     */
    private function cmdApiCacheCall(array $params) {
        HttpServer::request('ApiCache',[
            'method' => 'docking',
            'params'    => $params['params'],
           ],'127.0.0.1:' . HttpServer::serverPort());
        return true;
    }

    /**
     * 抛出异常
     * @param int $code 错误码
     * @throws Exception
     */
    private function throwErr($code = 0) {
        $errorList = [
            -3  => 'expired',
            -2  => 'ip limit',
            -1  => 'api Error',
            0   => 'params Error',
            1   => 'call huobi api error',
        ];
        throw new Exception($errorList[$code]);
    }


    /**
     * 获取服务器监听端口
     */
    static function serverPort()
    {
        if(!$port = env(static::ENV_KEY_LIST['SERVER_UDP_PORT'])) {
            $port = self::PROCESS_CONFIG['SERVER']['DEFAULT_PORT'];
        }
        return $port;
    }

    /**
     * 参数进行编码
     * @param $params 参数
     * @return string 内容
     */
    static function encode($params) {
        return base64_encode(json_encode($params));
    }

    /**
     * 解编
     * @param string $data
     */
    static function decode(string $data) {
        $result = null;
        try {
            $result = json_decode(base64_decode($data),true);
        }catch (\Exception $e) {
        }
        return $result;
    }


}